package com.cl.blog.web;

import com.alibaba.fastjson.JSONObject;
import com.cl.blog.common.Const;
import com.cl.blog.listener.RedisMessageListener2;
import com.cl.blog.util.JwtUtil;
import com.cl.blog.util.SpringUtils;
import io.jsonwebtoken.Claims;
import io.lettuce.core.ZStoreArgs;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.socket.WebSocketMessage;

import javax.servlet.http.HttpSession;
import javax.swing.*;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author 付
 * @Date 2024-03-27 13:30
 * @Version 1.0
 * @description
 * @since 1.0
 */
@ServerEndpoint("/socket/{userId}")
@RestController
@Slf4j
public class WebsocketEndpoint {
	/***
	 * 用来记录当前连接数的变量
	 */
	private static AtomicInteger onlineCount = new AtomicInteger(0);


	/***
	 * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象
	 */
//    private static CopyOnWriteArraySet<WebsocketEndpoint> webSocketSet = new CopyOnWriteArraySet<WebsocketEndpoint>();


	private static Map<String, Session> webSocketSet;
	//用户id
	private static String tokenid;
	private JwtUtil jwtUtil = SpringUtils.getBean(JwtUtil.class);
	/**
	 * 与某个客户端的连接会话，需要通过它来与客户端进行数据收发
	 */
	static private Session session;

	/**
	 * 得到线程池，执行并发操作
	 */
	private ThreadPoolTaskExecutor threadPoolTaskExecutor = SpringUtils.getBean(ThreadPoolTaskExecutor.class);


	//用来引入刚才在RedisConfig注入的类
	static private RedisMessageListenerContainer container = (RedisMessageListenerContainer) SpringUtils.getBean("container");


	// 自定义redis监听器
	static private RedisMessageListener2 listener2;

	static {
		webSocketSet = new ConcurrentHashMap<>();
	}

	/***
	 * socket打开的处理逻辑
	 * @throws Exception
	 */
	@OnOpen
	public void onOpen(Session session1, @PathParam("userId") String userId) throws Exception {

		session = session1;
		session.isOpen();
		tokenid = userId;
		//webSocketSet中存当前用户对象
		webSocketSet.put(session1.getId(), session1);
		//在线人数加一
		addOnlineCount();
		listener2 = new RedisMessageListener2();
		// 放入session
		listener2.setSession(session1);
		// 放入用户ID
//		listener.setUserId(userId);
		log.info(String.format("用户：%s 打开了Socket链接,$s", userId, session1.getId()));
		System.out.println("所有用户信息--->" + webSocketSet.toString());
		System.out.println("当前用户的信息--"+session1.getId());
		//初始化监听器
		container.addMessageListener(listener2, new PatternTopic(tokenid));
	}

	/**
	 * socket关闭的处理逻辑
	 */
	@OnClose
	public void onClose(Session session) {
		// 删除当前对象（this）
		webSocketSet.remove(session.getId());
		subOnlineCount();
		getOnlineCount();
		container.removeMessageListener(listener2);
		log.info(String.format("%s关闭了Socket链接Close a html ", tokenid));
	}


	/**
	 * socket收到消息的处理逻辑
	 */
	@OnMessage
	public void onMessage(String message, Session session) throws IOException {
		getOnlineCount();
		log.info("收到一条数据消息----------" + message + "---------" + session.getId() + "--" + webSocketSet.size());
		//可以自己根据业务处理
		for (String key : webSocketSet.keySet()) {
			if (webSocketSet.get(key).isOpen()) {
				// 使用线程池
				threadPoolTaskExecutor.execute(new Runnable() {
					@Override
					public void run() {
						// 使用锁，将当前的锁对象锁住
						synchronized (webSocketSet.get(key)) {
							System.out.println("用户信息————————" + webSocketSet.get(key).toString());
							try {
								if (session.getId() == key) {
									System.out.println("这里执行了");
									return;
								}
								webSocketSet.get(key).getBasicRemote().sendText(message);
							} catch (IOException e) {
								throw new RuntimeException(e);
							}
						}
					}
				});
			}
		}
	}

	/**
	 * 加一方法
	 */
	public volatile int p = 0;

	public synchronized void addOne() {
		p++;
		System.out.println(Thread.currentThread().getName() + "------->" + "自增==>" + p);
	}

	/**
	 * socket链接错误
	 *
	 * @param session
	 * @param error
	 */
	@OnError
	public void onError(Session session, Throwable error) {
		log.error("socket链接错误", error);
	}

	/**
	 * 发送消息
	 *
	 * @param message
	 */
	public void sendMessage(String message) throws IOException {
		if (session.isOpen()) {
			getOnlineCount();
			for (String key : webSocketSet.keySet()) {
				webSocketSet.get(key).getBasicRemote().sendText(message);
//				session.getBasicRemote().sendText(message);
			}
		}
	}


	/**
	 * 发送给指定用户
	 *
	 * @param message
	 * @param userid
	 */
	public void sendMessageToUser(String message, String userid) {
		System.out.println("发送的消息为："+message+"---接收的用户为："+userid);
		System.out.println(userid+"---"+webSocketSet.get(userid));
		if (userid != null) {
			try {
				webSocketSet.get(session.getId()).getBasicRemote().sendText(message);
			} catch (IOException e) {
				throw new RuntimeException(e);
			}
		}

	}


	//AtomicInteger是线程安全的 不需要synchronized修饰
	public static AtomicInteger getOnlineCount() {
		System.out.println(new Date() + "在线人数为" + onlineCount);
		return onlineCount;
	}

	//AtomicInteger是线程安全的 内置自增与自减的方法getAndIncrement()
	public static void addOnlineCount() {
		WebsocketEndpoint.onlineCount.getAndIncrement();
	}

	//AtomicInteger是线程安全的 内置自增与自减的方法getAndDecrement()
	public static void subOnlineCount() {
		WebsocketEndpoint.onlineCount.getAndDecrement();
	}

//	@Scheduled(fixedRate = 2000)
//	@Scheduled(fixedRate = 2000)
//	public void sendMes() throws IOException {
//		Claims claim = jwtUtil.getClaimByToken(this.tokenid, Const.GRADE_ADMIN);
//		//选择向所有人发送定时心跳
//		for (String key : webSocketSet.keySet()){
//			webSocketSet.get(key).sendMessage("心跳-->"+claim.getSubject());
//		}
//	}
}
